package io.github.eric.urpc.client;

import io.github.eric.urpc.client.handler.ClientMsgHandler;
import io.github.eric.urpc.client.handler.ClientConnectionHandler;
import io.github.eric.urpc.core.conn.Connection;
import io.github.eric.urpc.core.exception.UrpcRuntimeException;
import io.github.eric.urpc.core.rpc.Message;
import io.github.eric.urpc.core.rpc.Request;
import io.github.eric.urpc.core.rpc.Response;
import io.github.eric.urpc.core.rpc.RpcMappingResolver;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;

public class Client {
	private static final Logger LOGGER = LoggerFactory.getLogger(Client.class);

	private final Bootstrap bootstrap;
	private final EventLoopGroup workerGroup;
	private final ClientContext clientContext;
	private final ClientConnector clientConnector;

	public Client(ClientConfig clientConfig) {
		int cpuCore = Runtime.getRuntime().availableProcessors();
		this.clientContext = (ClientContext) clientConfig;
		this.clientContext.setLogger(LOGGER);
		if (clientContext.getMessageEncoder() == null) {
			throw new UrpcRuntimeException("[urpc client] message encoder is not present");
		}
		if (clientContext.getMessageDecoder() == null) {
			throw new UrpcRuntimeException("[urpc client] message decoder is not present");
		}
		if (clientContext.getRpcControllers() == null) {
			throw new UrpcRuntimeException("[urpc client] rpc controller is not present");
		}
		if (clientContext.getMessageRegistrar() == null) {
			throw new UrpcRuntimeException("[urpc client] message registrar is not present");
		}

		bootstrap = new Bootstrap();

		clientConnector = new ClientConnector(bootstrap, clientContext);
		clientContext.setConnectionManager(new ClientConnectionManager(this.clientContext, clientConnector));

		new RpcMappingResolver().resolve(this.clientContext);

		workerGroup = new NioEventLoopGroup(
				1, new DefaultThreadFactory(
						"urpc-client-" + clientConfig.getServerHost() + ":" + clientConfig.getServerPort() + "-worker-nioEventLoopGroup"));

		ClientMsgHandler clientMsgHandler = new ClientMsgHandler(bootstrap, clientContext);
		LengthFieldPrepender lengthFieldPrepender = new LengthFieldPrepender(4);
		final LoggingHandler loggingHandler;
		if (clientContext.getNettyLoggingLevel() != null) {
			loggingHandler = new LoggingHandler(clientContext.getNettyLoggingLevel());
		} else {
			loggingHandler = null;
		}

		bootstrap.group(workerGroup)
				.channel(NioSocketChannel.class)
				.option(ChannelOption.SO_KEEPALIVE, true)
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientContext.getSocketConnectTimeout()*1000)
				.handler(new ChannelInitializer<SocketChannel>() {
							@Override
							protected void initChannel(SocketChannel ch) throws Exception {
								if (loggingHandler != null) {
									ch.pipeline().addFirst(loggingHandler);
								}
								ch.pipeline().addLast(
										new IdleStateHandler(-1, clientContext.getHeartBeatInterval(), -1),
										new ClientConnectionHandler(bootstrap, clientContext),
										new LengthFieldBasedFrameDecoder(clientContext.getFrameMaxLength(),
										0, 4, 0,
										4),
										lengthFieldPrepender,
										// FOR TEST
//										new StringDecoder(),
//										new StringEncoder(),

										clientContext.getMessageEncoder(),
										clientContext.getMessageDecoder().copy(),
										clientMsgHandler
								);
							}
						});

		Thread shutdownHook = new Thread(new Thread(() -> {
			LOGGER.info("[urpc client] client shutdown hook started");
			shutdown();
			System.out.println("[urpc client] client shutdown hook ended");
			LOGGER.info("[urpc client] client shutdown hook ended");
		}),
		"urpc-client-" + clientConfig.getServerHost() + ":" + clientConfig.getServerPort() + "-shut-down-hook-thread");
		Runtime.getRuntime().addShutdownHook(shutdownHook);
	}

	public boolean isConnected() {
		Collection<Connection> connections = clientContext.getConnectionManager().getConnections();
        for (Connection connection : connections) {
            if (connection != null && connection.isConnected()) {
                return true;
            }
        }
		return false;
	}

	public void shutdown() {
		if (clientContext.isShuttingDown()) {
			return;
		}
		clientContext.shutdown();

		try {
			clientContext.getHeartBeatExecutor().shutdown();
		} catch (Exception e) {
			LOGGER.info("[urpc client] client heartbeat executor stop error.", e);
		}

		try {
			if (clientContext.getConnectionManager() != null) {
				clientContext.getConnectionManager().shutdown();
				LOGGER.info("[urpc client] client channels shut down done.");
			}
		} catch (Throwable t) {
			LOGGER.error("[urpc client] client channels shut down error", t);
		}

		try {
			if (workerGroup != null) {
				Future<?> future = workerGroup.shutdownGracefully().syncUninterruptibly();
				if (future.isSuccess()) {
					LOGGER.info("[urpc client] client worker group shut down successfully.");
				} else {
					LOGGER.warn("[urpc client] client worker group shut down done, there may be exception.");
				}
			}
		} catch (Throwable t) {
			LOGGER.error("[urpc client] client worker group shut down error", t);
		}

		try {
			if (clientContext.getHashedWheelTimer() != null) {
				clientContext.getHashedWheelTimer().stop();
				LOGGER.info("[urpc client] client hashed wheel timer stop.");
			}
		} catch (Exception e) {
			LOGGER.info("[urpc client] client hashed wheel timer stop error.", e);
		}

		try {
			clientContext.getRpcExecutor().shutdown();
		} catch (Exception e) {
			LOGGER.info("[urpc client] client rpc executor stop error.", e);
		}
	}

	public ClientConfig getClientConfig() {
		return clientContext;
	}

	public void sendAsync(Message message) {
		clientContext.getConnectionManager().sendAsync(message);
	}

	public <T extends Response> T sendSync(Request request) {
		return clientContext.getConnectionManager().sendSync(request);
	}
}
